package com.bfxy.rabbit.producer.broker;

import com.bfxy.rabbit.api.Message;
import com.bfxy.rabbit.api.MessageType;
import com.bfxy.rabbit.api.exception.MessageException;
import com.bfxy.rabbit.api.exception.MessageRuntimeException;
import com.bfxy.rabbit.common.convert.GenericMessageConverter;
import com.bfxy.rabbit.common.convert.RabbitMessageConverter;
import com.bfxy.rabbit.common.serializer.Serializer;
import com.bfxy.rabbit.common.serializer.SerializerFactory;
import com.bfxy.rabbit.common.serializer.impl.JacksonSerializerFactory;
import com.bfxy.rabbit.producer.service.MessageStoreService;
import com.google.common.base.Preconditions;
import com.google.common.base.Splitter;
import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;

/**
 * @describte:对RabbitTemplate进行池化操作，提高多生产者发送消息时的效率
 * @program:foodie-dev
 * @author:zhengdp
 * @create:2021-09-15 16:16:07
 **/
@Slf4j
@Component
public class RabbitTemplateContainer implements RabbitTemplate.ConfirmCallback{

    /**
     * Object ： ConcurrentMap 并发的Map
     * key : topic
     * value : RabbitTemplate
     */
    private Map<String, RabbitTemplate> rabbitMap = Maps.newConcurrentMap();

    // google的工具类
    private Splitter splitter = Splitter.on("#");

    // 序列化工厂
    private SerializerFactory serializerFactory = JacksonSerializerFactory.INSTANCE;

    @Autowired
    private MessageStoreService messageStoreService;

    // 连接工厂，直接注入即可
    @Autowired
    private ConnectionFactory connectionFactory;

    public RabbitTemplate getTemplate(Message message) throws MessageRuntimeException {
        Preconditions.checkNotNull(message);
        String topic = message.getTopic();
        RabbitTemplate rabbitTemplate = rabbitMap.get(topic);
        if(rabbitTemplate != null){
            return rabbitTemplate;
        }
        log.info("#RabbitTemplate.getTemplate# Containertopic : {} is not exists , create one",topic);
        // 当池中没有对应的RabbitTemplate，就新创建一个
        RabbitTemplate newRabbitTemplate = new RabbitTemplate(connectionFactory);
        newRabbitTemplate.setExchange(topic);
        // 重试时用到的
        newRabbitTemplate.setRetryTemplate(new RetryTemplate());
        // routingkey 设置
        newRabbitTemplate.setRoutingKey(message.getRoutingKey());

        //  对于Message的序列化方式,生成MessageConvert
        // 添加序列化反序列化和Converter对象
        Serializer serializer = serializerFactory.create();
        GenericMessageConverter gmc = new GenericMessageConverter(serializer);
        RabbitMessageConverter rmc = new RabbitMessageConverter(gmc);
         newRabbitTemplate.setMessageConverter(rmc);

        String messageType = message.getMessageType();
        if(!MessageType.RAPID.equals(messageType)){
            // 只有迅速消息不需要设置callback
            newRabbitTemplate.setConfirmCallback(this);
        }

        // 如果不存在才进行put操作
        rabbitMap.putIfAbsent(topic,newRabbitTemplate);

        return rabbitMap.get(topic);
    }

    /**
     * 由 RabbitTemplate容器来实现ConfirmCallBack方法
     * 除了迅速消息，其他消息都可以对消息的应答进行一个callback
     * @param correlationData
     * @param ack
     * @param cause
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        // 具体的消息应答

        // 按#分割coorelationData得到MessageId
        List<String> strings = splitter.splitToList(correlationData.getId());
        String messageId = strings.get(0);
        long sendTime = Long.parseLong(strings.get(1));
        String messageType = strings.get(2);

        // broker成功收到消息
        if(ack){
            // 当Broker返回ACK成功时
            if(MessageType.Reliant.endsWith(messageType)){
                // 如果消息类型为Reliant的话，就更新数据库中的Msg日志记录
                this.messageStoreService.success(messageId);
            }
            log.info("send message is OK, confirm messageId: {},snedTime: {}",messageId,sendTime);
        }else{
            // broker 接收消息失败
            // 根据不同的失败原因，进行不同的补偿策略
            log.info("send message is Fail, confirm messageId: {},snedTime: {}",messageId,sendTime);
        }
    }
}
